package io.plus.engine.pulse.task;

import io.plus.engine.client.ClientAsynchronousSocketChannelContext;
import io.plus.engine.client.ClientConfiguration;
import io.plus.engine.client.ReConnectConfiguration;
import io.plus.engine.pulse.AsynchronousSocketChannelContext;
import io.plus.engine.pulse.Configuration;
import io.plus.engine.pulse.task.queue.Queue;
import io.plus.engine.pulse.task.queue.QueueImpl;
import io.plus.engine.utils.lock_capable_task.AbstractQueueTask;
import lombok.extern.slf4j.Slf4j;
import tool.plus.Clock;

import java.util.concurrent.Executor;

@Slf4j
public class CloseTask extends AbstractQueueTask<AsynchronousSocketChannelContext> {

    private Queue<AsynchronousSocketChannelContext> queue = null;

    public CloseTask(Executor executor) {
        super(executor);
        getQueue();
    }

    /**
     * 彻底删除，不再维护
     */
    public static void remove(AsynchronousSocketChannelContext asynchronousSocketChannelContext) {

        Configuration configuration = asynchronousSocketChannelContext.clientConfiguration;
        if (!configuration.isServer()) {
            /**
             * 客户端
             */
            ClientConfiguration clientConfiguration = (ClientConfiguration) configuration;
            clientConfiguration.closeds.remove(asynchronousSocketChannelContext);
            clientConfiguration.connecteds.remove(asynchronousSocketChannelContext);
        }

        configuration.connections.remove(asynchronousSocketChannelContext);
        configuration.ips.unbind(asynchronousSocketChannelContext);
        configuration.ids.unbind(asynchronousSocketChannelContext);

        unbind(asynchronousSocketChannelContext);
    }

    public static void unbind(AsynchronousSocketChannelContext asynchronousSocketChannelContext) {
        Configuration configuration = asynchronousSocketChannelContext.clientConfiguration;
        configuration.users.unbind(asynchronousSocketChannelContext);
        configuration.groups.unbind(asynchronousSocketChannelContext);
        //上面两个比较重要所以放在一堆
        configuration.tokens.unbind(asynchronousSocketChannelContext);
        configuration.businessIds.unbind(asynchronousSocketChannelContext);
    }

    @Override
    public Queue<AsynchronousSocketChannelContext> getQueue() {
        if (queue == null) {
            synchronized (this) {
                if (queue == null) {
                    queue = new QueueImpl<AsynchronousSocketChannelContext>();
                }
            }
        }
        return queue;
    }

    @Override
    public void runTask() {
        if (queue.isEmpty()) {
            return;
        }
        AsynchronousSocketChannelContext asynchronousSocketChannelContext = null;
        while ((asynchronousSocketChannelContext = queue.poll()) != null) {
            try {
                boolean isNeedRemove = asynchronousSocketChannelContext.closeTag.isNeedRemove;
                String remark = asynchronousSocketChannelContext.closeTag.remark;
                Throwable throwable = asynchronousSocketChannelContext.closeTag.throwable;
                //
                asynchronousSocketChannelContext.asynchronousSocketChannelState.closedTime = Clock.currentTimeMillis;

                try {
                    if (asynchronousSocketChannelContext.isClosed && !isNeedRemove) {
                        log.info("{}, {}已经关闭，备注:{}，异常:{}", asynchronousSocketChannelContext.clientConfiguration, asynchronousSocketChannelContext, remark, throwable == null ? "无" : throwable.toString());
                        return;
                    }
                    if (asynchronousSocketChannelContext.isRemoved) {
                        log.info("{}, {}已经删除，备注:{}，异常:{}", asynchronousSocketChannelContext.clientConfiguration, asynchronousSocketChannelContext, remark, throwable == null ? "无" : throwable.toString());
                        return;
                    }
                    //必须先取消任务再清空队列
                    asynchronousSocketChannelContext.decodeTask.setCanceled(true);
                    asynchronousSocketChannelContext.handlerTask.setCanceled(true);
                    asynchronousSocketChannelContext.sendTask.setCanceled(true);

                    asynchronousSocketChannelContext.decodeTask.clearQueue();
                    asynchronousSocketChannelContext.handlerTask.clearQueue();
                    asynchronousSocketChannelContext.sendTask.clearQueue();
                    //
                    log.info("{}, {} 准备关闭连接, isNeedRemove:{}, {}", asynchronousSocketChannelContext.clientConfiguration, asynchronousSocketChannelContext, isNeedRemove, remark);
                    try {
                        if (isNeedRemove) {
                            remove(asynchronousSocketChannelContext);
                        } else {
                            ClientConfiguration clientConfiguration = (ClientConfiguration) asynchronousSocketChannelContext.clientConfiguration;
                            clientConfiguration.closeds.add(asynchronousSocketChannelContext);
                            clientConfiguration.connecteds.remove(asynchronousSocketChannelContext);
                            unbind(asynchronousSocketChannelContext);
                        }
                        asynchronousSocketChannelContext.setRemoved(isNeedRemove);
                        //
                        asynchronousSocketChannelContext.asynchronousSocketChannelState.closedTime = Clock.currentTimeMillis;
                        asynchronousSocketChannelContext.setClosed(true);
                    } catch (Throwable e) {
                        log.error(e.toString(), e);
                    } finally {
                        if (!isNeedRemove && asynchronousSocketChannelContext.isClosed && !asynchronousSocketChannelContext.isServer()) //不删除且没有连接上，则加到重连队列中
                        {
                            ClientAsynchronousSocketChannelContext clientChannelContext = (ClientAsynchronousSocketChannelContext) asynchronousSocketChannelContext;
                            ReConnectConfiguration.put(clientChannelContext);
                        }
                    }
                } catch (Throwable e) {
                    log.error(throwable.toString(), e);
                }

            } finally {
                asynchronousSocketChannelContext.isWaitingClose = false;
            }
        }
    }


}
